package com.haizhi.messagehub.service.redis;

import java.util.Map;

import javax.annotation.Resource;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.DependsOn;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.stereotype.Service;

import com.haizhi.messagehub.bean.form.ConsumerAdapterForm;
import com.haizhi.messagehub.bean.form.MessageForm;
import com.haizhi.messagehub.service.MessageHubServiceImpl;

@Service("redisPubSubProcessor")
public class RedisPubSubProcessor extends MessageHubServiceImpl {

    @Resource
    Map<String, MessageListenerAdapter> redisPubSubConsumerMap;

    @Override
    public void producer(MessageForm messageForm) {
        stringRedisTemplate.convertAndSend(messageForm.getTopic(), messageForm.getMessage());
    }

    @Override
    public void consumer(ConsumerAdapterForm messageForm) {
        MessageListenerAdapter adapter = new MessageListenerAdapter(messageForm.getBean(), messageForm.getInvokeMethod().getName());
        adapter.afterPropertiesSet();

        redisPubSubConsumerMap.put(messageForm.getTopic(), adapter);
    }


    @Bean
    public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) {

        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);

        redisPubSubConsumerMap.forEach((topic, adapter) -> {

            container.addMessageListener(adapter, new PatternTopic(topic));
        });
        return container;
    }
}
